package com.bw.gmall.realtime.db;

import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdShop_Pay {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        String topic = "dwd_traffic_page_log_a";
        String groupId = "dws_traffic_page";

        // 添加时间生成和水位线
        tEnv.executeSql("" +
                "create table if not exists page_log( " +
                "    `common` map<string, string>, " +
                "    `page` map<string, string>, " +
                "    `ts` bigint, " +
                "    `rt` as TO_TIMESTAMP(FROM_UNIXTIME(ts / 1000)), " +
                "    WATERMARK FOR rt AS rt - INTERVAL '2' SECOND " +
                " ) " + MyKafkaUtil.getKafkaDDL(topic, groupId));

        // 转换行时间属性为普通时间戳
        Table pageTable = tEnv.sqlQuery(" select " +
                " common['shop_id'] shop_id," +
                " common['mid_id'] mid_id," +
                " common['uid'] uid, " +
                " CAST(rt AS TIMESTAMP) AS rt_new " +
                " from page_log "
        );
        tEnv.createTemporaryView("page_table", pageTable);

        tEnv.executeSql(MyKafkaUtil.getTopicDb("topic_db"));

        // 订单表、订单详情表、店铺表
        Table orderTable = tEnv.sqlQuery(" select \n" +
                " CAST(data['id'] AS BIGINT)  id, \n" +
                " CAST(data['user_id'] AS BIGINT)  user_id, \n" +
                " CAST(data['total_amount'] AS DECIMAL(16, 2)) total_amount, \n" +
                " CAST(data['activity_reduce_amount'] AS DECIMAL(16, 2)) activity_reduce_amount, \n" +
                " CAST(data['coupon_reduce_amount'] AS DECIMAL(16, 2)) coupon_reduce_amount, \n" +
                " data['order_status'] order_status, \n" +
                " CAST(data['create_time'] AS TIMESTAMP(3)) order_create_time \n" +
                "from topic_db \n" +
                "where `database` = 'gmall'\n" +
                "and `table` = 'order_info' " +
                "and  (`type` = 'insert' or `type` = 'update')");
        tEnv.createTemporaryView("order_info", orderTable);

        Table orderDileTable = tEnv.sqlQuery("select \n" +
                " CAST(data['order_id'] AS BIGINT)  order_id, \n" +
                " CAST(data['sku_id'] AS BIGINT)  sku_id, \n" +
                " CAST(data['sku_num'] AS BIGINT) sku_num, \n" +
                " CAST(data['order_price'] AS DECIMAL(10, 2)) order_price\n" +
                "from topic_db \n" +
                "where `database` ='gmall'\n" +
                "and `table` = 'order_detail'" +
                " and  (`type` = 'insert' or `type` = 'update')");
        tEnv.createTemporaryView("order_detail", orderDileTable);

        Table shopTable = tEnv.sqlQuery("select \n" +
                " CAST(data['sku_id'] AS BIGINT)  sku_id, \n" +
                " data['shop_id']  shop_id \n" +
                "from topic_db\n" +
                "where `database` ='gmall'\n" +
                "and `table` = 'shop_sku'");
        tEnv.createTemporaryView("shop_sku", shopTable);

        Table py_table = tEnv.sqlQuery(" SELECT \n" +
                "    s.shop_id,\n" +
                "    COUNT(p.mid_id) AS mid_count,\n" +
                "    COUNT(DISTINCT p.uid) AS uid_count,\n" +
                "    COALESCE(SUM(i.total_amount), 0) AS money,\n" +
                "    COALESCE(COUNT(DISTINCT i.user_id), 0) AS user_count,\n" +
                "    COALESCE(\n" +
                "    COUNT(DISTINCT i.user_id) / NULLIF(COUNT(DISTINCT p.mid_id), 0), \n" +
                "        0\n" +
                "    ) * 100 AS pays\n" +
                "FROM shop_sku s\n" +
                "LEFT JOIN page_table p ON s.shop_id = p.shop_id\n" +
                "JOIN order_detail d ON s.sku_id = d.sku_id\n" +
                "JOIN order_info i ON d.order_id = i.id \n" +
                " where i.order_status = '1004' " +
                "GROUP BY s.shop_id");
        tEnv.createTemporaryView("py_table", py_table);
//        tEnv.executeSql("select * from py_table ").print();

        tEnv.executeSql(" create table page_pay( " +
                " shop_id varchar(255), " +
                " mid_count bigint, " +
                " uid_count bigint, " +
                " money DECIMAL(16, 2), " + // -- 与 SUM(i.total_amount) 类型一致
                " user_count bigint, " +
                " pays double, " +
                " PRIMARY KEY (shop_id) NOT ENFORCED " + // 定义逻辑主键（假设根据业务 shop_id 可作为主键，可根据实际调整）
                " ) " + MyKafkaUtil.getUpsertKafkaDDL("page_pay"));

        // 执行插入并输出结果
        tEnv.executeSql("insert into page_pay select * from py_table");

        tEnv.executeSql("select * from page_pay").print();




        env.execute("DwdShop_Pay Job");
    }
}